package com.itbaizhan.rocketmqproducer.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

//顺序消费
public class OrderConsumer {
    public static void main(String[] args) throws MQClientException {
        //创建消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_grp_01");
        //设置nameserver地址
        consumer.setNamesrvAddr("192.168.66.100:9876");
        //订阅的主题  与生产端一至    不匹配标签 *
        consumer.subscribe("topic_springboot_01","*");
        //最小线程数  因为是按顺序消费所以设置1
        consumer.setConsumeThreadMin(1);
        //最大线程数
        consumer.setConsumeThreadMax(1);
        //一次拉取的数量 1
        consumer.setPullBatchSize(1);
        //一次消费的消息数量
        consumer.setConsumeMessageBatchMaxSize(1);

        //使用有序的消息监听器消费消息  实现接口MessageListenerOrderly()
        consumer.setMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                for(MessageExt msg :list){
                    System.out.println(msg.getTopic()+"\t"+
                            msg.getQueueId()+"\t"+
                            new String(msg.getBody()));
                }
                //返回状态
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
    }

}
